// Copyright 2013 Petar Maymounkov
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package ripple

import (
	"io"
	"math/rand"
	"runtime"
	"sync"

	"circuit/kit/rh"
)

//
type transaction interface {

	// Wait blocks until a lock can be obtained, the transaction is interrupted, aborted or the lock is reset.
	// Nil error signifies the lock was acquited.
	// io.ErrNoProgress signifies the wait was interrupted.
	// io.ErrUnexpectedEOF signifies the lock was reset during the wait.
	// io.ErrClosedPipe signifies the transaction is already the current holder of the lock.
	Wait(rh.Intr) error

	// Clunk aborts the transaction, unblocking any pending calls to Wait
	Clunk()
}

//
type tx struct {
	r struct {
		ch <-chan surrenderFunc
		sync.Mutex
		surr surrenderFunc
	}
	s struct {
		sync.Mutex
		ch chan<- surrenderFunc
	}
}

type surrenderFunc func(*tx)

func (t *tx) Wait(intr rh.Intr) error {
	t.r.Lock()
	defer t.r.Unlock()
	//
	if t.r.surr != nil {
		return io.ErrClosedPipe // Already holding lock
	}
	//
	select {
	case surr, ok := <-t.r.ch:
		if !ok {
			return io.ErrUnexpectedEOF
		}
		t.r.surr = surr
		//
		return nil
	case <-intr:
		return io.ErrNoProgress
	}
}

func (t *tx) Clunk() {
	t.abort(true)
}

// abort closes out the transaction and, if surrender is set, calls into the surrender function, if any.
func (t *tx) abort(surrender bool) {
	t.s.Lock() // Close send side
	if t.s.ch != nil {
		close(t.s.ch)
		t.s.ch = nil
	}
	t.s.Unlock()
	//
	t.r.Lock() // Close receive side
	if surrender && t.r.surr != nil {
		t.r.surr(t)
	}
	t.r.surr = nil
	t.r.Unlock()
}

func (t *tx) unblock(surr surrenderFunc) {
	t.s.Lock()
	defer t.s.Unlock()
	if t.s.ch == nil { // If no-one is waiting; transaction is closed or abandoned
		surr(t)
		return
	}
	t.s.ch <- surr
	close(t.s.ch)
	t.s.ch = nil
}

//
type Semaphore struct {
	sync.Mutex
	waiting map[*tx]struct{}
	holding *tx // Transaction, currently holding the lock
}

func (s *Semaphore) Init() {
	s.waiting = make(map[*tx]struct{})
}

func (s *Semaphore) NewTx() transaction {
	s.Lock()
	defer s.Unlock()
	//
	var ch = make(chan surrenderFunc, 1)
	t := &tx{}
	t.r.ch = ch
	t.s.ch = ch
	runtime.SetFinalizer(t, func(x *tx) {
		x.Clunk()
	})
	//
	if s.holding == nil { // Lock not held
		s.holding = t
		go t.unblock(s.surrender)
		return t
	}
	s.waiting[t] = struct{}{}
	return t
}

func (s *Semaphore) NumWaiting() int {
	s.Lock()
	defer s.Unlock()
	//
	return len(s.waiting)
}

func (s *Semaphore) Stat() (nholding, nwaiting int) {
	s.Lock()
	defer s.Unlock()
	//
	if s.holding != nil {
		nholding = 1
	}
	return nholding, len(s.waiting)
}

func (s *Semaphore) Reset() {
	s.Lock()
	defer s.Unlock()
	//
	if s.holding != nil {
		s.holding.abort(false)
	}
	s.holding = nil
	//
	for t, _ := range s.waiting {
		t.abort(false)
	}
	s.waiting = make(map[*tx]struct{})
}

// surrender is closed by the transaction object to notify the Semaphore that its user is done using it.
func (s *Semaphore) surrender(t *tx) {
	s.Lock()
	defer s.Unlock()
	//
	if s.holding != t && s.holding != nil { // Lock already held by someone else. Nothing to do.
		return
	}
	s.holding = nil
	// Pick a random waiter to unblock
	n := len(s.waiting)
	if n == 0 {
		return
	}
	for {
		for t, _ := range s.waiting {
			if rand.Intn(n) != 0 {
				continue
			}
			delete(s.waiting, t)
			go t.unblock(s.surrender)
			return
		}
	}
}
